package com.mango.iot.gateway.tcp.mq.consumer;

import com.mango.iot.gateway.tcp.connection.Connection;
import com.mango.iot.gateway.tcp.connection.ConnectionManager;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * 给设备发送下行通知 消费者
 *
 * @author liangfeihu
 * @since 2023/6/27 16:53
 */
@Slf4j
@Component
public class DeviceInstructionConsumer {

    @Autowired
    private ConnectionManager connectionManager;


    @KafkaListener(topics = "iot_tcp_service")
    public void process(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object tcpMessage = kafkaMessage.get();
            log.info("消费者收到消息  : " + tcpMessage.toString());

            //tcpMessage.get("deviceId").toString();
            String deviceId = "";
            Connection conn = connectionManager.getConn(deviceId);
            if (conn == null) {
                log.warn("[InstructionSend]deviceId={} this pc 无链接或该设备已下线", deviceId);
                return;
            }
            ChannelHandlerContext ctx = conn.getCtx();
            if (ctx.channel().isWritable()) {
                byte[] setBytes = null;
                // convert tcpMessage
                ctx.pipeline().writeAndFlush(Unpooled.wrappedBuffer(setBytes));
            } else {
                log.info("[InstructionSend]tcp channel {} not writable", deviceId);
            }
        }
    }

}
